package cn.initcap;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author initcap
 * @date Created in 9/3/19 9:25 AM.
 */
public class WindowsReduceJava {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> text = env.socketTextStream("localhost", 9999);

        text.flatMap((String value, Collector<Tuple2<Integer, Integer>> out) -> {
            String[] tokens = value.toLowerCase().split(",");

            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(1, Integer.parseInt(token)));
                }
            }
        }).keyBy(0)
                .timeWindow(Time.seconds(5))
                .reduce((Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) -> {
                    System.out.println("value1 = [" + value1 + "], value2 = [" + value2 + "]");
                    return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
                })
                .print()
                .setParallelism(1);


        env.execute("WindowsReduceJava");

    }

}
